package com.navinfo.opentsp.platform.computing.serializer;

import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.RealTimeDataPb;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;

/**
 * 0F37实时统计
 * HDFS_Sink的RealTimeData序列化器，拼接成字段间通过\001分割的字符串
 */
public class HDFSRealTimeSerializer implements EventSerializer {
        private static final Logger logger = LoggerFactory.getLogger(HDFSRealTimeSerializer.class);

        private final String APPEND_NEWLINE = "appendNewline";
        private final boolean appendNewline;
        private final String splitStr = "\001";
        private final OutputStream out;

        private HDFSRealTimeSerializer(OutputStream out, Context ctx) {
            this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
            this.out = out;
        }

    public boolean supportsReopen() {
        return false;
    }

    public void afterCreate() {}

    public void afterReopen() {}

    public void beforeClose() {}

    public void flush() throws IOException {
        this.out.flush();
    }

    public void write(Event event) throws IOException {
        try {
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            RealTimeDataPb.RealTimeData realTimeData = RealTimeDataPb.RealTimeData.parseFrom(event.getBody());

            StringBuilder locationStr = new StringBuilder(10000);
            locationStr
                    .append(realTimeData.getTerminalId()).append(splitStr)
                    .append(realTimeData.getGpsTime()).append(splitStr)
                    .append(realTimeData.getLatitude()).append(splitStr)
                    .append(realTimeData.getLongitude()).append(splitStr)
                    .append(realTimeData.getHeight()).append(splitStr)
                    .append(realTimeData.getEngineOutputTorque()).append(splitStr)
                    .append(realTimeData.getSpeed()).append(splitStr)
                    .append(realTimeData.getAccelerator()).append(splitStr)
                    .append(realTimeData.getBrake()).append(splitStr)
                    .append(realTimeData.getRotation()).append(splitStr)
                    .append(realTimeData.getGear()).append(splitStr)
                    .append(realTimeData.getClutchSwitch()).append(splitStr)
                    .append(realTimeData.getRealTimeOilConsumption()).append(splitStr)
                    .append(realTimeData.getFuelConsumptionRate()).append(splitStr)
                    .append(realTimeData.getVersion()).append(splitStr)
                    .append(realTimeData.getRetransmissionPackets()).append(splitStr)
                    .append(realTimeData.getSerialNumber()).append(splitStr)
                    .append(realTimeData.getReceiveDate()).append(splitStr)
                    .append(realTimeData.getOriginalLat()).append(splitStr)
                    .append(realTimeData.getOriginalLng()).append(splitStr)
                    .append(realTimeData.getProtocolType()).append(splitStr)
                    .append(realTimeData.getVehicleSpeed()).append(splitStr)
                    .append(realTimeData.getBarometricPressure()).append(splitStr)
                    .append(realTimeData.getFrictionTorque()).append(splitStr)
                    .append(realTimeData.getScrUpStreamNOxSensorOutput()).append(splitStr)
                    .append(realTimeData.getScrDownStreamNOxSensorOutput()).append(splitStr)
                    .append(realTimeData.getReagentAllowance()).append(splitStr)
                    .append(realTimeData.getAirInflow()).append(splitStr)
                    .append(realTimeData.getScrInletTemperature()).append(splitStr)
                    .append(realTimeData.getScrOutletTemperature()).append(splitStr)
                    .append(realTimeData.getDpfDropoutVoltage()).append(splitStr)
                    .append(realTimeData.getEngineCoolantTemperature()).append(splitStr)
                    .append(realTimeData.getTankLevel()).append(splitStr)
                    .append(realTimeData.getPositioningState()).append(splitStr)
                    .append(realTimeData.getMileage()).append(splitStr)
                    .append(realTimeData.getRValue()).append(splitStr)
                    .append(realTimeData.getSValue()).append(splitStr)
                    .append(realTimeData.getCommandUnit()).append(splitStr)
                    .append(realTimeData.getVin()).append(splitStr)
                    .append(realTimeData.getTerminalVersion()).append(splitStr)
                    .append(realTimeData.getEncryption()).append(splitStr)
                    .append(realTimeData.getRLength()).append(splitStr)
                    .append(realTimeData.getSLength()).append(splitStr)
                    .append(realTimeData.getEngineBrakingState()).append(splitStr)
                    .append(realTimeData.getExhaustBrakingState()).append(splitStr)
                    .append(realTimeData.getEmptySignal()).append(splitStr)

//                    .append(realTimeData.getAuxiliaryBrakeHandleStatus()).append(splitStr)

                    .append(realTimeData.getExhaustFlow()).append(splitStr)
                    .append(realTimeData.getFanSpeed()).append(splitStr)
                    .append(realTimeData.getMechanicalBoxGear()).append(splitStr)
                    .append(realTimeData.getInstrumentSpeed()).append(splitStr)
                    .append(realTimeData.getWheelSpeed()).append(splitStr)
                    .append(realTimeData.getParkingBrakeSwitch()).append(splitStr)
                    .append(realTimeData.getCruiseControlSettingSwitch()).append(splitStr)
                    .append(realTimeData.getTargetGear()).append(splitStr)
                    .append(realTimeData.getSlope()).append(splitStr)
                    .append(realTimeData.getLoad()).append(splitStr)
                    .append(realTimeData.getFuelLevel()).append(splitStr)
                    .append(realTimeData.getWaterTemperature()).append(splitStr)
                    .append(realTimeData.getAtmosphericPressure()).append(splitStr)
                    .append(realTimeData.getIntakeAirTemperature()).append(splitStr)
                    .append(realTimeData.getAtmosphericTemperature()).append(splitStr)
                    .append(realTimeData.getExhaustGasTemperature()).append(splitStr)
                    .append(realTimeData.getIntakeManifoldBoostPressure()).append(splitStr)
                    .append(realTimeData.getRelativePressure()).append(splitStr)
                    .append(realTimeData.getEngineTorqueMode()).append(splitStr)
                    .append(realTimeData.getOilPressure()).append(splitStr)
                    .append(realTimeData.getUreaLevel()).append(splitStr)
                    .append(realTimeData.getStatusFlag()).append(splitStr)
                    .append(realTimeData.getBrakeRate()).append(splitStr)
                    .append(realTimeData.getGpsDirection()).append(splitStr)
                    .append(realTimeData.getAirConditionStatus()).append(splitStr)
                    .append(realTimeData.getTransmissionRotation()).append(splitStr)
                    .append(realTimeData.getAuxiliaryBrakeHandleStatus()).append(splitStr)
                    .append(realTimeData.getUreaInjection()).append(splitStr)
                    .append(realTimeData.getTotalureaInjection()).append(splitStr)
                    .append(realTimeData.getUreaInjectionSignal()).append(splitStr)
                    .append(realTimeData.getUreaPumpPressureSignal()).append(splitStr)
                    .append(realTimeData.getAvgGasConsumption()).append(splitStr)
                    //添加字段只能在分割线之上添加
                    //------------------------------------------分割线----------------------------------------
                    .append(DateUtils.format(realTimeData.getGpsTime(), "yyyyMMdd")).append(splitStr)
                    .append(realTimeData.getTerminalId()%32);


            //将0F37报文拼接后写入hdfs
            this.out.write(locationStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的RealTime拦截器失败：" + e.getMessage(), e);
        }
    }

    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSRealTimeSerializer(out, context);
        }
    }
}
